/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package hadoop

import org.apache.hadoop.io.BytesWritable
import org.apache.hadoop.mapreduce.Mapper
import parser.Nodes._
import parser.DataTypes._
import com.google.protobuf.DescriptorProtos._
import com.google.protobuf.Descriptors._
import com.google.protobuf.ByteString
import util.PBUtils
import pbdata.Tables.KeyValue
import generator.Emitter
import generator.Helpers

class GenMapper extends Mapper[BytesWritable, BytesWritable, BytesWritable, BytesWritable] {

  type Context = Mapper[BytesWritable, BytesWritable, BytesWritable, BytesWritable]#Context

  var mapper: Helpers.Mapper = null
  var nTables = 0

  override def setup(context: Context): Unit = {
    val conf = context.getConfiguration
    val c = SCConfiguration(conf)
    val fds = PBUtils.convertFiles(c.protoFiles).values.toList

    mapper  = generator.Main.newMapper(fds, c.staticVariables)
    nTables = c.tables.size
  }

  class SingleEmitter(context: Context) extends Emitter {
    override def write(id: Int, key: ByteString, value: ByteString): Unit = {
      // Ignore ID
      context.write(
        new BytesWritable(key.toByteArray), 
        new BytesWritable(value.toByteArray))
    }
  }

  class MultiEmitter(context: Context) extends Emitter {
    override def write(id: Int, key: ByteString, value: ByteString): Unit = {
      context.write(new BytesWritable(id.toString.getBytes("UTF-8")),
                    new BytesWritable(KeyValue.newBuilder.setKey(key)
                                                         .setValue(value).build.toByteArray))
    }
  }

  override def map(key: BytesWritable, value: BytesWritable, context: Context): Unit = {
    val emitter = nTables match {
      case 0 => new SingleEmitter(context) // Dummy, not used.
      case 1 => new SingleEmitter(context)
      case _ => new MultiEmitter(context)
    }

    mapper.map(emitter, new Helpers.ByteStringWrapper(ByteString.copyFrom(value.getBytes(), 0, value.getLength)))
  }
}
